package com.sh.data.engine.domain.dataquality.service.impl;

import com.alibaba.fastjson.JSONArray;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.sh.data.engine.common.constants.HttpStatus;
import com.sh.data.engine.common.constants.code.BizCodes;
import com.sh.data.engine.common.enumDefinition.DSType;
import com.sh.data.engine.common.enumDefinition.DataQualityRuleCompareTypeEnum;
import com.sh.data.engine.common.exception.BusinessException;
import com.sh.data.engine.common.util.DateUtil;
import com.sh.data.engine.common.util.TableNameUtil;
import com.sh.data.engine.domain.base.model.UserContext;
import com.sh.data.engine.domain.common.service.CommonService;
import com.sh.data.engine.domain.dataquality.model.domain.DataQualityRuleTemplateDomain;
import com.sh.data.engine.domain.dataquality.model.domain.RuleCondition;
import com.sh.data.engine.domain.dataquality.service.*;
import com.sh.data.engine.domain.dataquality.utils.DataQualityUtils;
import com.sh.data.engine.domain.shims.db.BaseDbManager;
import com.sh.data.engine.domain.shims.sql.builder.QueryBuilder;
import com.sh.data.engine.domain.util.ConvertUtil;
import com.sh.data.engine.domain.util.LogUtil;
import com.sh.data.engine.domain.util.VarParserUtil;
import com.sh.data.engine.domain.workspace.index.service.ProjectService;
import com.sh.data.engine.repository.dao.dataquality.entity.*;
import com.sh.data.engine.repository.dao.workspace.index.entity.ProjectEntity;
import com.sh.data.engine.repository.dao.workspace.manager.entity.StorageEntity;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hive.jdbc.HiveStatement;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import static com.sh.data.engine.domain.dataquality.utils.DataQualityUtils.*;

/**
 * @author tinglan.ys
 * @description:
 * @date 2022/2/25 下午2:14
 */
@Service
@Slf4j
public class DataQualityJobExecServiceImpl implements DataQualityJobExecService {

    @Lazy
    @Autowired
    private DataQualityService dataQualityService;

    @Autowired
    private DataQualityRuleTemplateService dataQualityRuleTemplateService;

    @Autowired
    private DataQualityTablePartitionService dataQualityTablePartitionService;

    @Autowired
    private DataQualityJobService dataQualityJobService;

    @Autowired
    private DataQualityJobRuleService dataQualityJobRuleService;

    @Autowired
    private DataQualityJobRuleFieldService dataQualityJobRuleFieldService;

    @Lazy
    @Autowired
    private CommonService commonService;

    @Lazy
    @Autowired
    private ProjectService projectService;

    private static final String TOKEN_APP = "The url to track the job: ";

    private static final Pattern P_APP = Pattern.compile("/proxy/([^/]+)");

    private ExecutorService executor;

    @PostConstruct
    private void init() {
        // 定义线程池
        executor =
                new ThreadPoolExecutor(
                        50,
                        50,
                        0L,
                        TimeUnit.MILLISECONDS,
                        new LinkedBlockingQueue<>(10),
                        Executors.defaultThreadFactory(),
                        new ThreadPoolExecutor.AbortPolicy());
    }

    @Override
    public void executeDataqualityJob(
            Long projectId,
            String tableName,
            DataQualityJobEntity dataqualityJob,
            List<DataQualityRuleEntity> rules) {

        // 定义执行任务需要的参数。key为规则id，value为模版、自定义规则、分区表达式
        Map<Long, DataQualityRuleTemplateEntity> ruleTemplates = Maps.newHashMap();
        Map<Long, String> diyTemplates = Maps.newHashMap();
        Map<Long, String> partitions = Maps.newHashMap();

        for (DataQualityRuleEntity rule : rules) {
            // 如果是模板规则 查询模板
            if (rule.getRuleDiy() == null || rule.getRuleDiy().equals(0)) {
                DataQualityRuleTemplateEntity template =
                        dataQualityRuleTemplateService.getById(rule.getTemplateId());
                ruleTemplates.put(rule.getId(), template);
            } else {
                // 如果是自定义模板 新逻辑是语句自定义
                //        String sql =
                //            String.format(
                //                "select %s(${field}) from %s", rule.getRuleDiyStatisticalModel(),
                // tableName);
                String sql = rule.getDiySql();
                diyTemplates.put(rule.getId(), sql);
            }

            // 如果有分区查询分区 并解析分区
            if (rule.getPartitionId() != null) {
                DataQualityTablePartitionEntity dataqualityTablePartition =
                        dataQualityTablePartitionService.getById(rule.getPartitionId());
                // 分区解析:如果查到了分区表达式而且分区表达式不是"NOPARTITIONTABLE"，存入分区表达式
                if (StringUtils.isNotBlank(dataqualityTablePartition.getPartitionExpress())
                        && !"NOPARTITIONTABLE"
                        .equalsIgnoreCase(dataqualityTablePartition.getPartitionExpress())) {
                    String partition =
                            VarParserUtil.getDataqualityPt(dataqualityTablePartition.getPartitionExpress());
                    partitions.put(rule.getId(), partition);
                }
            }
        }
        String userId = UserContext.getUserId();
        DataQualityRunnable task =
                new DataQualityRunnable(
                        projectId, tableName, dataqualityJob, rules, ruleTemplates, diyTemplates, partitions, userId);
        executor.submit(task);
    }

    @Data
    class DataQualityRunnable implements Runnable {

        private Long projectId;

        private String tableName;

        private DataQualityJobEntity dataqualityJob;

        private List<DataQualityRuleEntity> rules;

        private Map<Long, DataQualityRuleTemplateEntity> templates;
        private Map<Long, String> diyTemplates;
        private Map<Long, String> partitions;
        private String userId;

        public DataQualityRunnable(
                Long projectId,
                String tableName,
                DataQualityJobEntity dataqualityJob,
                List<DataQualityRuleEntity> rules,
                Map<Long, DataQualityRuleTemplateEntity> templates,
                Map<Long, String> diyTemplates,
                Map<Long, String> partitions,
                String userId) {
            this.projectId = projectId;
            this.tableName = tableName;
            this.dataqualityJob = dataqualityJob;
            this.rules = rules;
            this.templates = templates;
            this.diyTemplates = diyTemplates;
            this.partitions = partitions;
            this.userId = userId;
        }

        @Override
        public void run() {
            // 持久化执行记录
            dataQualityService.insertNewestJob(dataqualityJob, userId);
            // 根据不同的存储，替换表名
            StorageEntity storageEntity = commonService.getStorage(projectId);
            String storageType = storageEntity.getStorageType();

            // 如果是pg或者sap 表名要替换为"schema"."tableName",加入引号是因为要适配大小写。
            if (storageType.equals(DSType.PostgreSQL.name())
                    || storageType.equals(DSType.MatrixDB.name())
                    || storageType.equals(DSType.Sap.name()) || storageType.equals(DSType.KINGBASE8)) {
                tableName = TableNameUtil.getTblNameLong(tableName);
            } else if (storageType.equals(DSType.DM.name())) {
                String dbName = storageEntity.getRmdbs().getDbName();
                tableName = "\"" + dbName + "\"." + "\"" + tableName + "\"";
            }

            int errorNum = 0;
            int sqlErrorNum = 0;
            try {

                for (DataQualityRuleEntity rule : rules) {
                    try {
                        // 执行每一个规则
                        execute(rule, storageType);
                    } catch (BusinessException e) {
                        // 0-运行中 1-正常 2-异常 3-sql语句出错
                        if (BizCodes.DATA_QUALITY_RULE_EXCEPTION_ERROR.getCode() == e.getCode()) {
                            errorNum++;
                        } else if (BizCodes.DATA_QUALITY_SQL_STATEMENT_ERROR.getCode() == e.getCode()) {
                            sqlErrorNum++;
                        }

                        // 如果强类型规则 直接跳出循环不再执行
                        if (rule.getRuleStrength().equals(0)) {
                            break;
                        }
                    }
                }
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            } finally {
                // 关闭连接资源

                // 根据实际情况设置完成状态
                DataQualityJobEntity dataqualityJob = new DataQualityJobEntity();
                dataqualityJob.setErrorNum(errorNum + sqlErrorNum);
                if (errorNum == 0 && sqlErrorNum == 0) {
                    dataqualityJob.setStatus(1);
                } else if (sqlErrorNum > 0 && errorNum == 0) {
                    dataqualityJob.setStatus(3);
                } else if (errorNum > 0) {
                    dataqualityJob.setStatus(2);
                }
                dataqualityJob.setUpdateTime(new Date());
                dataqualityJob.setId(this.dataqualityJob.getId());
                dataQualityJobService.updateById(dataqualityJob);
            }
        }

        private void execute(DataQualityRuleEntity rule, String storageType) {
            // 生成规则执行记录
            DataQualityJobRuleEntity dataqualityJobRule = new DataQualityJobRuleEntity();
            dataqualityJobRule.setCreateTime(new Date());
            dataqualityJobRule.setDataqualityJobId(dataqualityJob.getId());
            dataqualityJobRule.setRuleId(rule.getId());
            dataqualityJobRule.setRuleName(rule.getRuleName());
            dataqualityJobRule.setRuleFrom(rule.getRuleFrom());
            dataqualityJobRule.setRuleCompareType(rule.getRuleCompareType());
            dataqualityJobRule.setRuleExpectValue(rule.getRuleExpectValue());
            dataqualityJobRule.setStartTime(new Date());
            dataqualityJobRule.setStatus(0);
            dataQualityJobRuleService.save(dataqualityJobRule);

            // 根据规则拼写sql sqlTemp为要执行的语句
            String sqlTemp;
            String partition = partitions.get(rule.getId());
            DataQualityRuleTemplateDomain dataQualityRuleTemplateDomain =
                    ConvertUtil.copyProperties(
                            templates.get(rule.getId()), DataQualityRuleTemplateDomain.class);

            String[] fields = {"-"};
            String[] fieldTypes = {"-"};

            if (rule.getRuleDiy().equals(1)) {
                // 自定义规则
                StringBuilder sb = new StringBuilder();
                sb.append(diyTemplates.get(rule.getId()));
                if (StringUtils.isNotBlank(partition)) {
                    sb.append(" ").append("where").append(" ").append(partition);
                }
                sqlTemp = sb.toString();
                // 自定义规则，初始化规则字段和规则类型（无规则字段和规则类型）
            } else {
                // 根据不同的存储拿到对应的模版sql 替换sql中的table
                sqlTemp =
                        DataQualityUtils.replaceTableParam(
                                dataQualityRuleTemplateDomain, partition, tableName, storageType);

                // 模版规则，初始化规则字段和规则类型
                fields = rule.getFieldNames().split(",");
                fieldTypes = rule.getFieldTypes().split(",");
            }

            // sql都没有直接报错
            if (StringUtils.isBlank(sqlTemp)) {
                dataqualityJobRule.setEndTime(new Date());
                dataqualityJobRule.setErrorNum(0);
                dataqualityJobRule.setStatus(3);
                dataQualityJobRuleService.updateById(dataqualityJobRule);
                throw new BusinessException(BizCodes.DATA_QUALITY_SQL_STATEMENT_ERROR, HttpStatus.BAD_REQUEST);
            }

            int errorNum = 0;
            int sqlErrorNum = 0;

            // 挨个执行字段
            for (int i = 0; i < fields.length; i++) {

                // 替换字段和其他条件 生成最终可执行的sql
                String execSql =
                        DataQualityUtils.getExecSql(
                                fields[i],
                                fieldTypes[i],
                                sqlTemp,
                                rule.getRuleCompareType(),
                                rule.getRuleCondition(),
                                dataQualityRuleTemplateDomain,
                                storageType,
                                rule.getTemplateFilter());

                // sql执行结果
                String value = "";

                // 生成每个字段的执行记录
                DataQualityJobRuleFieldEntity dataqualityJobRuleField = new DataQualityJobRuleFieldEntity();
                dataqualityJobRuleField.setDataqualityJobRuleId(dataqualityJobRule.getId());
                dataqualityJobRuleField.setFieldName(fields[i]);
                dataqualityJobRuleField.setStatus(0);
                dataqualityJobRuleField.setCreateTime(new Date());
                dataQualityJobRuleFieldService.save(dataqualityJobRuleField);

                try {
                    // 执行sql并获取执行后值

                    log.info("数据质量任务执行sql:" + execSql);
                    value = doSql(execSql, storageType);
                    log.info("数据质量任务执行结果:" + value);

                    // 根据规则检查
                    Boolean flag = DataQualityUtils.checkRule(rule, value);

                    if (!flag) {
                        throw new BusinessException(BizCodes.DATA_QUALITY_RULE_EXCEPTION_ERROR, HttpStatus.BAD_REQUEST);
                    }

                    dataqualityJobRuleField.setStatus(1);
                } catch (BusinessException e) {
                    //  0-运行中 1-正常 2-异常 3-sql语句出错
                    if (BizCodes.DATA_QUALITY_RULE_EXCEPTION_ERROR.getCode() == e.getCode()) {
                        dataqualityJobRuleField.setStatus(2);
                        errorNum++;
                    } else if (BizCodes.DATA_QUALITY_SQL_STATEMENT_ERROR.getCode() == e.getCode()) {
                        dataqualityJobRuleField.setStatus(3);
                        sqlErrorNum++;
                        // 如果强类型规则 直接跳出循环不再执行
                        if (rule.getRuleStrength().equals(0)) {
                            break;
                        }
                    }
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                    dataqualityJobRuleField.setStatus(2);
                    errorNum++;
                    // 如果强类型规则 直接跳出循环不再执行
                    if (rule.getRuleStrength().equals(0)) {
                        break;
                    }
                } finally {
                    // dataqualityJobRuleField.setUpdateTime(new Date());

                    ProjectEntity projectById = projectService.getProjectById(projectId);

                    if (dataqualityJobRuleField.getStatus() == 2
                            || dataqualityJobRuleField.getStatus() == 3) {
                        dataqualityJobRuleField.setWarnContent(
                                this.formatWarnContent(
                                        execSql,
                                        value,
                                        tableName,
                                        projectById.getName(),
                                        dataqualityJobRuleField.getFieldName(),
                                        dataqualityJobRuleField.getStatus(),
                                        rule));
                    }

                    dataQualityJobRuleFieldService.updateById(dataqualityJobRuleField);
                }
            }

            dataqualityJobRule.setEndTime(new Date());
            dataqualityJobRule.setErrorNum(sqlErrorNum + errorNum);
            if (errorNum == 0 && sqlErrorNum == 0) {
                dataqualityJobRule.setStatus(1);
            } else if (sqlErrorNum > 0 && errorNum == 0) {
                dataqualityJobRule.setStatus(3);
            } else if (errorNum > 0) {
                dataqualityJobRule.setStatus(2);
            }
            dataqualityJobRule.setUpdateTime(new Date());
            dataQualityJobRuleService.updateById(dataqualityJobRule);
            if (!dataqualityJobRule.getStatus().equals(1)) {
                if (dataqualityJobRule.getStatus().equals(3)) {
                    throw new BusinessException(BizCodes.DATA_QUALITY_SQL_STATEMENT_ERROR, HttpStatus.BAD_REQUEST);
                } else {
                    throw new BusinessException(BizCodes.DATA_QUALITY_RULE_EXCEPTION_ERROR, HttpStatus.BAD_REQUEST);
                }
            }
        }

        private String doSql(String execSql, String storageType) {

            CountDownLatch latch = new CountDownLatch(1);

            BaseDbManager dbManager = commonService.getStorageDbManagerByProjctId(projectId);
            QueryBuilder queryBuilder = null;

            ResultSet rs = null;

            try {

                if (storageType.equals(DSType.Hive.name())) {
                    List<String> yarnJobIds = Lists.newArrayList();
                    HiveStatement stmt = (HiveStatement) dbManager.getConnection().createStatement();
                    processLogHive(stmt, latch, yarnJobIds);

                    boolean hasRS = stmt.execute(execSql);
                    if (hasRS) {
                        rs = stmt.getResultSet();
                    }
                } else {
                    // 除了hive的其他数据源 pg sap tidb
                    queryBuilder = dbManager.getQueryBuilder();
                    rs = queryBuilder.selectSql(execSql).list();
                }

                String value = null;
                if (rs.next()) {
                    value = rs.getString(1);
                }
                return value;

            } catch (Exception e) {
                log.error(e.getMessage(), e);
                throw new BusinessException(BizCodes.DATA_QUALITY_SQL_STATEMENT_ERROR, HttpStatus.BAD_REQUEST);
            } finally {
                if (null != dbManager) {
                    dbManager.close();
                }
                if (rs != null) {
                    try {
                        rs.close();
                    } catch (SQLException throwables) {
                        log.error(throwables.getMessage(), throwables);
                    }
                }
                if (storageType.equals(DSType.Hive.name())) {
                    await(latch);
                } else {
                    queryBuilder.close();
                }
            }
        }

        private void processLogPG(Statement stmt, CountDownLatch latch) {
            new Thread(
                    () -> {
                        try {

                            // PG的日志代码，目前不支持此功能
                        } catch (Exception e) {
                            log.error("日志读取失败{}", e.getMessage());
                        } finally {
                            latch.countDown();
                        }
                    })
                    .start();
        }

        private void processLogHive(HiveStatement stmt, CountDownLatch latch, List<String> yarnJobIds) {
            new Thread(
                    () -> {
                        try {
                            while (!stmt.isClosed() && stmt.hasMoreLogs()) {
                                List<String> logList = stmt.getQueryLog();
                                if (CollectionUtils.isEmpty(logList)) {
                                    try {
                                        TimeUnit.MICROSECONDS.sleep(100);
                                    } catch (Exception ignore) {
                                    }
                                    continue;
                                }
                                // 更新yarn_job_id
                                for (String log : logList) {
                                    if (StringUtils.contains(log, TOKEN_APP)) {
                                        Matcher m = P_APP.matcher(log);
                                        if (m.find()) {
                                            String jobId = m.group(1);
                                            yarnJobIds.add(jobId);
                                        }
                                    }
                                }
                                LogUtil.writeDataqualityLog(dataqualityJob.getId(), logList);
                            }
                        } catch (Exception e) {
                            log.error("日志读取失败{}", e.getMessage());
                        } finally {
                            latch.countDown();
                        }
                    })
                    .start();
        }

        private void await(CountDownLatch latch) {
            try {
                latch.await();
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        }

        private String formatWarnContent(
                String execSql,
                String value,
                String table,
                String projectName,
                String fieldName,
                Integer status,
                DataQualityRuleEntity rule) {

            StringBuilder content = new StringBuilder();
            content
                    .append("[")
                    .append(DateUtil.formatDate(new Date()))
                    .append("]")
                    .append(" 空间为:")
                    .append(projectName)
                    .append(";表名为:")
                    .append(table)
                    .append(";字段为:")
                    .append(fieldName)
                    .append("规则名为:")
                    .append(rule.getRuleName())
                    .append("出现:");

            if (status == 3) {
                content.append("sql语句出错:" + execSql);
                return content.toString();
            }

            content.append("数据质量问题:");
            String compareType = "";
            if (StringUtils.isNotBlank(rule.getRuleCompareType())
                    && !"null".equals(rule.getRuleCompareType())) {
                if (rule.getRuleCompareType().split(",").length == 1) {
                    if (null
                            != DataQualityRuleCompareTypeEnum.getByCompareType(
                            Integer.valueOf(rule.getRuleCompareType()))) {
                        compareType =
                                DataQualityRuleCompareTypeEnum.getByCompareType(
                                                Integer.valueOf(rule.getRuleCompareType()))
                                        .getDesc();
                    }
                }
            }

            String msg = "";

            if (rule.getRuleDiy() == 0) {
                DataQualityRuleTemplateEntity ruleTemplate = templates.get(rule.getId());
                String min = "";
                String max = "";
                switch (String.valueOf(ruleTemplate.getId())) {
                    case WZXYS:
                        msg = String.format("唯一值记录数和总记录数的比值%s%s", compareType, rule.getRuleExpectValue());
                        break;
                    case KZYS:
                        msg = "存在空值";
                        break;
                    case KZZBYS:
                        msg = String.format("空值占比%s%s", compareType, rule.getRuleExpectValue());
                        break;
                    case QZFWYS:
                        // min = this.getCondition(rule.getRuleCondition(), "range");
                        // max = this.getCondition(rule.getRuleCondition(), "range");
                        // msg = String.format("取值范围%s[%s,%s]", compareType, min, max);
                        msg = String.format("取值范围规则不匹配");
                        break;
                    case ZDZYS:
                        max = this.getCondition(rule.getRuleCondition(), "max");
                        msg = String.format("最大值大于%s", max);
                        break;
                    case ZXZYS:
                        min = this.getCondition(rule.getRuleCondition(), "min");
                        msg = String.format("最小值小于%s", min);
                        break;
                    case PJZYS:
                        String avg = this.getCondition(rule.getRuleCondition(), "avg");
                        msg = String.format("平均值%s%s", compareType, avg);
                        break;
                    case ZZYS:
                        String sum = this.getCondition(rule.getRuleCondition(), "sum");
                        msg = String.format("总值%s%s", compareType, sum);
                        break;
                    case CDYS:
                        // String length = this.getCondition(rule.getRuleCondition(), "length");
                        // msg = String.format("最大长度值%s%s", compareType, length);
                        msg = String.format("长度约束规则不匹配");
                        break;
                    case DMCDYS:
                        msg = String.format("非指定长度的记录数%s%s", compareType, rule.getRuleExpectValue());
                        break;
                    case ZZBDSYS:
                        msg = "存在非身份证号格式的值";
                        break;
                    case RQZYS:
                        String date = this.getCondition(rule.getRuleCondition(), "date");
                        msg = String.format("日期值%s%s", compareType, date);
                        break;
                    case MJZYS:
                        String enumVal = this.getCondition(rule.getRuleCondition(), "enum");
                        msg = String.format("枚举值%s[%s]", compareType, enumVal);
                        break;
                    case KZFCYS:
                        msg = "值存在空字符串";
                        break;
                    case KZHZFCYS:
                        msg = "值存在空值或空字符串";
                        break;
                    case TSZFYS:
                        String specialChar = this.getCondition(rule.getRuleCondition(), "specialChar");
                        msg = String.format("特殊字符%s与规则匹配", specialChar);
                        break;
                    case CFZYS:
                        msg = "存在重复值";
                        break;
                    case JSXYST:
                        msg = "数据在天级别存在延迟";
                        break;
                    case ZZBDSYSDH:
                        msg = "存在非电话号码格式的值";
                        break;
                    case JSXYSX:
                        msg = "数据在小时级别存在延迟";
                        break;
                    default:
                }
            } else {
                //        msg =
                //            String.format(
                //                "%s值%s%s",
                //                rule.getRuleDiyStatisticalModel(), compareType,
                // rule.getRuleExpectValue());
            }
            content.append(msg);
            content.append(";执行的sql为:").append(execSql).append(";结果为:").append(value);

            return content.toString();
        }

        private String getCondition(String ruleCondition, String key) {
            try {
                if (StringUtils.isNotBlank(ruleCondition)) {
                    List<RuleCondition> ruleConditions =
                            JSONArray.parseArray(ruleCondition, RuleCondition.class);

                    if (CollectionUtils.isNotEmpty(ruleConditions)) {
                        Optional<RuleCondition> optional =
                                ruleConditions.stream()
                                        .filter(c -> StringUtils.equalsIgnoreCase(c.getKey(), key))
                                        .findFirst();
                        if (optional.isPresent()) {
                            return optional.get().getValue();
                        }
                    }
                }
            } catch (Exception ignore) {
            }
            return null;
        }
    }
}
